﻿using Confluent.Kafka;
using Leo.Common.MQ;
using Leo.MSATracert;

namespace Leo.Common.Kafka
{
    public class KafkaConsumer : MQConsumerBase
    {
        public KafkaConsumer(KafkaConsumerConfig kcc, CancellationToken cancellationToken, params IMessageHandler[] handles) : base((handles != null) ? Enumerable.ToList<IMessageHandler>(handles) : null)
        {
            this.config = kcc;
            this.cancellationToken = cancellationToken;
        }

        protected override void ConnectToListen()
        {
            IConsumer<Ignore, string> consumer = new ConsumerBuilder<Ignore, string>(new ConsumerConfig
            {
                GroupId = this.config.GroupId,
                ClientId = this.config.ClientId,
                BootstrapServers = this.config.Broker,
                AutoOffsetReset = AutoOffsetReset.Earliest,
            }).Build();
            consumer.Subscribe(this.config.Topic);
            this.StartConsumeAsync(consumer);
        }

        private void StartConsumeAsync(IConsumer<Ignore, string> consumer)
        {
            Task.Factory.StartNew(delegate ()
            {
                while (!this.cancellationToken.IsCancellationRequested)
                {
                    try
                    {
                        this.MessageReceiveAndHandle(consumer);
                    }
                    catch (Exception ex)
                    {
                        Console.WriteLine("接收kafka消息出现异常，错误信息：" + ex.Message);
                        Console.WriteLine(ex.StackTrace);
                    }
                }
            }, this.cancellationToken, TaskCreationOptions.LongRunning, TaskScheduler.Current);
        }

        private void MessageReceiveAndHandle(IConsumer<Ignore, string> consumer)
        {
            ConsumeResult<Ignore, string> consumeResult = consumer.Consume(this.cancellationToken);
            if (consumeResult.IsPartitionEOF)
            {
                Console.WriteLine(string.Format("Reached end of topic {0}, partition {1}, offset {2}.", consumeResult.Topic, consumeResult.Partition, consumeResult.Offset));
                return;
            }
            string messageType = consumeResult.TryGetHeaderValue("_t_");
            string correlationId = consumeResult.TryGetHeaderValue("CorrelationId");
            string traceId = consumeResult.TryGetHeaderValue("traceId") ?? Guid.NewGuid().ToString("N").ToLower();
            MessageData message = new MessageData
            {
                Content = consumeResult.Message.Value,
                MessageType = messageType,
                CorrelationId = correlationId,
                MessageId = consumeResult.TryGetHeaderValue("MessageId"),
                QueueType = MessageQueueType.Kafka
            };
            ServiceInvocationLinkTrace.GetPrincipal(traceId, null);
            this.HandleMessage(message);
        }

        private Task HandleMessage(MessageData message)
        {
            List<IMessageHandler> matchHandlers = base.GetMatchHandlers(message.MessageType);
            if (matchHandlers == null || !matchHandlers.Any<IMessageHandler>())
            {
                return Task.CompletedTask;
            }
            if (matchHandlers.Count != 1)
            {
                return this.MultiExecute(matchHandlers, message);
            }
            return this.SingleExecute(matchHandlers.First<IMessageHandler>(), message);
        }

        private Task<bool> SingleExecute(IMessageHandler handler, MessageData message)
        {
            IMessageHandlerV2 messageHandlerV = handler as IMessageHandlerV2;
            if (messageHandlerV != null)
            {
                return messageHandlerV.Handle(message);
            }
            return handler.Handle(message.Content);
        }

        private Task MultiExecute(List<IMessageHandler> handlers, MessageData message)
        {
            foreach (var handler in handlers)
            {
                IMessageHandlerV2 messageHandlerV = handler as IMessageHandlerV2;
                if (messageHandlerV != null)
                {
                    return messageHandlerV.Handle(message);
                }
                return handler.Handle(message.Content);
            }
            return Task.CompletedTask;
        }

        private readonly KafkaConsumerConfig config;

        private readonly CancellationToken cancellationToken;
    }
}
